package com.atguigu.gmallrealtime.app.dim;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmallrealtime.app.func.DimConnectStreamBroadcastFunc;
import com.atguigu.gmallrealtime.app.func.DimSinkFunc;
import com.atguigu.gmallrealtime.bean.TableProcess;
import com.atguigu.gmallrealtime.common.Constant;
import com.atguigu.gmallrealtime.util.HBaseUtil;
import com.atguigu.gmallrealtime.util.JDBCUtil;
import com.atguigu.gmallrealtime.util.MyKafkaUtil;
import com.atguigu.gmallrealtime.util.MysqlUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;

/**
 * @author yhm
 * @create 2023-09-22 10:14
 */
public class DimApp {
    public static void main(String[] args) throws Exception {
        // TODO 1 创建flink环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        // TODO 2 添加检查点和状态后端
//        env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE);
//
//        //2.2 设置检查点超时时间
//        env.getCheckpointConfig().setCheckpointTimeout(60000L);
//        //2.3 设置job取消之后 检查点是否保留
//        env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//        //2.4 设置两个检查点之间最小的时间间隔
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000L);
//        //2.5 设置重启策略
//        // env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,3000L));
//        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30), Time.seconds(3)));
//
//        env.setStateBackend(new HashMapStateBackend());
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
//
//        System.setProperty("HADOOP_USER_NAME","atguigu");

        // TODO 3 读取kafka的ods层数据 topic_db
        String topicName = Constant.TOPIC_ODS_DB;
        String groupId = "dim_app_group";
        DataStreamSource<String> kafkaSource = env.fromSource(MyKafkaUtil.getKafkaSource(topicName, groupId), WatermarkStrategy.noWatermarks(), "kafka_source");

//        kafkaSource.print("kafka_source");

        // TODO 4 简单清洗数据 转换数据结构
        // 原始数据为maxwell监控的所有数据  有一些不需要
        // 类型由于不同表 无法统一 先使用jsonObject
        SingleOutputStreamOperator<JSONObject> filterSource = kafkaSource.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                try {
                    JSONObject jsonObject = JSON.parseObject(value);
                    // maxwell数据的类型type -> insert update delete bootstrap-start bootstrap-insert bootstrap-complete
                    String type = jsonObject.getString("type");
                    if (!"bootstrap-start".equals(type) && !"bootstrap-complete".equals(type)) {
                        out.collect(jsonObject);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });

//        filterSource.print("filter");

        // TODO 5 flinkCDC读取配置表(记录哪些是维度表)信息
        DataStreamSource<String> flinkCDCSource = env.fromSource(MysqlUtil.getMySqlSource("gmall_config", "gmall_config.table_process"), WatermarkStrategy.noWatermarks(), "table_process").setParallelism(1);

//        flinkCDCSource.print("flinkCDC");

        // TODO 6 转换配置表数据结构
        SingleOutputStreamOperator<TableProcess> processStream = flinkCDCSource.process(new ProcessFunction<String, TableProcess>() {
            @Override
            public void processElement(String value, Context ctx, Collector<TableProcess> out) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                String op = jsonObject.getString("op");
                TableProcess tableProcess = null;

                if ("d".equals(op)) {
                    tableProcess = JSON.parseObject(jsonObject.getString("before"), TableProcess.class);
                } else {
                    tableProcess = JSON.parseObject(jsonObject.getString("after"), TableProcess.class);

                }
                tableProcess.setOp(op);

                // 到对应的hbase当中创建表格
                if ("d".equals(op)) {
                    // 删除对应表格
                    HBaseUtil.dropTable(Constant.HBASE_NAMESPACE, tableProcess.getSinkTable());
                } else if ("u".equals(op)) {
                    // 先删除before
                    HBaseUtil.dropTable(Constant.HBASE_NAMESPACE, jsonObject.getJSONObject("before").getString("sink_table"));
                    // 再创建after
                    HBaseUtil.createTable(Constant.HBASE_NAMESPACE, tableProcess.getSinkTable(), tableProcess.getSinkFamily());
                } else {
                    // 创建表格
                    HBaseUtil.createTable(Constant.HBASE_NAMESPACE, tableProcess.getSinkTable(), tableProcess.getSinkFamily());
                }

                out.collect(tableProcess);
            }
        });
//        processStream.print("tableProcess");


        // TODO 7 连接主流和配置流
        // key(sourceTable+sourceType) value(tableProcess)
        MapStateDescriptor<String, TableProcess> tableProcessDesc = new MapStateDescriptor<>("table_process", String.class, TableProcess.class);
        BroadcastStream<TableProcess> processBroadcastStream = processStream.broadcast(tableProcessDesc);
        BroadcastConnectedStream<JSONObject, TableProcess> connectStream = filterSource.connect(processBroadcastStream);

        // TODO 8 判断筛选主流中的维度表数据
        // 对连接流进行处理 需要3个泛型(主流类型  广播流类型  返回值类型)
        SingleOutputStreamOperator<Tuple2<TableProcess, JSONObject>> connectProcessStream = connectStream.process(new DimConnectStreamBroadcastFunc(tableProcessDesc));

//        connectProcessStream.print("connectProcess");


        // TODO 9 筛选出需要的字段
        // 筛选的标志是table_process中的sink_columns和type
        SingleOutputStreamOperator<Tuple2<TableProcess, JSONObject>> Tuple2Stream = connectProcessStream.map(new MapFunction<Tuple2<TableProcess, JSONObject>, Tuple2<TableProcess, JSONObject>>() {
            @Override
            public Tuple2<TableProcess, JSONObject> map(Tuple2<TableProcess, JSONObject> value) throws Exception {
                JSONObject jsonObject = value.f1;
                TableProcess tableProcess = value.f0;
                List<String> stringList = Arrays.asList(tableProcess.getSinkColumns().split(","));
                JSONObject data = jsonObject.getJSONObject("data");

                data.keySet().removeIf(key -> !stringList.contains(key));
                data.put("op_type", jsonObject.getString("type"));
                return Tuple2.of(tableProcess, data);
            }
        });

//        Tuple2Stream.print("Tuple2Stream");

        // TODO 10 写出到hbase
        Tuple2Stream.addSink(new DimSinkFunc());

        // TODO 11 执行环境
        env.execute();
    }
}
